package com.hhf.rrd.wordcount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;

/**
 * 词频统计
 *
 * @author huanghaifeng15
 * @date 2022/2/11 09:53
 **/
public class WordCountApp2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        List<String> data = Arrays.asList("spark,spark,spark", "flink,flink");

        environment.fromCollection(data)
                .flatMap((String line, Collector<String> outer) -> {
                    Arrays.stream(line.split(",")).forEach(outer::collect);
                }).returns(Types.STRING)
                .map(x -> Tuple2.of(x, 1))
                // 如果函数使用泛型返回，flink无法进行推断，则需要根据该方法为算子进行指定。
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(new KeySelector<Tuple2<String,Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> input) throws Exception {
                        return input.f0;
                    }
                })
                .sum(1).print();

        environment.execute();
    }
}
